package com.rr.streamrabbitmqnew.service;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

import java.util.function.Supplier;
import java.util.stream.Stream;

/**
 * @author zhuangcanrun
 * @Date 2021/06/17/10:05
 * @description
 **/

@Service
public class SendService {
    @Autowired
    private StreamBridge streamBridge;

    public void sendMessageToSend(String message){
        System.out.println("发送消息：{" + message + "}到send-in-0");
        streamBridge.send("enrichecho-in-0", message);
    }

    public void send(String destinationAndMessage) {
        String[] split = destinationAndMessage.split(",");
        System.out.println("发送消息：{" + split[1] + "}到{" + split[0] + "}");
        streamBridge.send(split[0], split[1]);
    }

    @Bean
    public Supplier<Flux<String>> stringSupplier() {
        return () -> Flux.fromStream(Stream.generate(new Supplier<String>() {
            @Override
            public String get() {
                try {
                    Thread.sleep(1000);
                    return "Hello from Supplier";
                } catch (Exception e) {
                    // ignore
                    return "";
                }
            }
        })).subscribeOn(Schedulers.elastic()).share();
    }

}
